package cn.monkey.transport.core;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Collection;
import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

public interface DelegatesExchange extends Exchange {
    Collection<Exchange> delegates();

    @Override
    default Inbound inbound() {
        throw new UnsupportedOperationException("delegates exchange has no inbound");
    }

    @Override
    default Outbound outbound() {
        throw new UnsupportedOperationException("delegates exchange has no outbound");
    }

    @Override
    default Mono<Void> write(Object o) {
        List<Mono<Void>> list = this.delegates().stream().map(delegate -> delegate.write(o)).collect(Collectors.toList());
        return Flux.merge(list).then();
    }

    @Override
    default Mono<Void> flush() {
        List<Mono<Void>> list = this.delegates().stream().map(Exchange::flush).collect(Collectors.toList());
        return Flux.merge(list).then();
    }

    @Override
    default Mono<Void> writeAndFlush(Object o) {
        List<Mono<Void>> list = this.delegates().stream().map(exchange -> exchange.writeAndFlush(o)).collect(Collectors.toList());
        return Flux.merge(list).then();
    }

    @Override
    default boolean isActive() {
        return this.delegates().stream().anyMatch(Exchange::isActive);
    }

    @Override
    default <T> T setAttribute(String key, T val) {
        return this.delegates().stream()
                .map(exchange -> exchange.setAttribute(key, val))
                .filter(Objects::nonNull)
                .findAny()
                .orElse(null);
    }

    @Override
    @SuppressWarnings("unchecked")
    default <T> T getAttribute(String key) {
        return this.delegates().stream()
                .map(exchange -> {
                    try {
                        return (T) exchange.getAttribute(key);
                    } catch (ClassCastException e) {
                        return null;
                    }
                })
                .filter(Objects::nonNull)
                .findAny().orElse(null);
    }

    @Override
    default Mono<Void> close() {
        List<Mono<Void>> list = this.delegates().stream().map(Exchange::close).collect(Collectors.toList());
        return Flux.merge(list).then();
    }

    default Mono<Void> destroy(Exchange exchange) {
        if (!this.delegates().contains(exchange)) {
            return Mono.error(new UnsupportedOperationException("exchange not exists"));
        }
        return exchange.destroy();
    }

    @Override
    default Exchange bind() {
        this.delegates().forEach(Exchange::bind);
        return this;
    }
}
